/**
 * @作者 伟子涵
 * @微信公号 搬砖暗夜码农
 * @开源项目 $ https://gitee.com/zhouwei5208888
 * @我的博客 $ http://itzhouwei.com
 * @创建时间 2023/9/27 14:57
 */
package com.itzhouwei.rpc.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;


import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @description: 连接管理器  单列的
 * @author：zw，微信：yingshengzw
 * @date: 2023/9/27
 * @Copyright： 公众号：搬砖暗夜码农 | 博客：https://itzhouwei.com - 沉淀、分享、成长，让自己和他人都能有所收获！
 */
@Slf4j
@SuppressWarnings("all")
public class RpcConnectManager {


    private static volatile RpcConnectManager RPC_CONNECT_MANAGER = new RpcConnectManager();

    private RpcConnectManager() {
    }


    //缓存  一个连接的地址 对应一个实际的业务处理器
    private Map<InetSocketAddress, RpcClientHandler> concurrentHashMap = new ConcurrentHashMap<InetSocketAddress, RpcClientHandler>();

    /* 所有连接成功的地址 所对应的 任务执行器列表 connectedHandlerList   */
    private CopyOnWriteArrayList<RpcClientHandler> connectedHandlerList = new CopyOnWriteArrayList<RpcClientHandler>();

    /* 异步线程池 用于异步提交连接的线程池 */
    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(16, 16, 60, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(65536));

    /* 锁 */
    private ReentrantLock reentrantLock=new ReentrantLock();

    private Condition condition=reentrantLock.newCondition();

    private EventLoopGroup eventLoopGroup = new NioEventLoopGroup(4);


    /**超时时间 等于6秒 **/
   private Long connectTimeoutMills=6000L;


   private volatile boolean  isRunning=true;

   private volatile AtomicInteger handlerIdx=new AtomicInteger(0);

    /***
     * 单例模式
     * @return
     */
    public static RpcConnectManager getInstance() {
        return RPC_CONNECT_MANAGER;
    }

    //1. 异步创建连接  线程池  真正的发起连接  连接失败监听  连接成功监听

    /****
     * <h1>发起连接方法 </h1>
     * @param serverAddress
     */
    public void connect(final String serverAddress) {
        List<String> allServerAddress = Arrays.asList(serverAddress.split(","));
        updateConnectedServer(allServerAddress);
    }

    /****
     * <h1>更新缓存信息 并异步发起连接</h1>
     * 192.168.11.111:8765,192.168.11.112:8759
     * @param allServerAddress
     */
    public void updateConnectedServer(List<String> allServerAddress) {
        if (CollectionUtils.isNotEmpty(allServerAddress)) {

            //为了简单 这里使用本地缓存
            HashSet<InetSocketAddress> newAllServerAddress = new HashSet<>();
            for (int i = 0; i < allServerAddress.size(); i++) {
                String[] array = allServerAddress.get(i).split(":");
                if (array.length == 2) {
                    String host = array[0];
                    int port = Integer.parseInt(array[1]);
                    final InetSocketAddress remotePeer = new InetSocketAddress(host, port);
                    newAllServerAddress.add(remotePeer);
                }
            }
            //2 调用建立连接方法  发起远程连接操作
            for (InetSocketAddress serverNodeAddress : newAllServerAddress) {
                //缓存里面没有的 我在去发起异步连接
                if (!concurrentHashMap.keySet().contains(serverNodeAddress)) {
                    connectAsync(serverNodeAddress);
                }
            }
            //3 TODO allServerAddress列表里不存在的连接地址  那么我需要从缓存中进行移除
            for (int i = 0; i<connectedHandlerList.size(); i++) {
                RpcClientHandler rpcClientHandler = connectedHandlerList.get(i);
                SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
                if (!newAllServerAddress.contains(remotePeer)){
                    log.info("删除 旧的连接地址 "+remotePeer);
                    //删除和释放资源
                    RpcClientHandler handler = concurrentHashMap.get(remotePeer);
                    if (handler!=null) {
                        handler.close();
                        concurrentHashMap.remove(remotePeer);
                        connectedHandlerList.remove(rpcClientHandler);
                    }

                }
            }

        } else {
            //添加警告
            log.error("no available server address!");
            //清除所有的缓存信息
            clearConnected();
        }
    }

    /***
     * <h1>异步发起连接的方法</h1>
     * @param serverNodeAddress
     */
    private void connectAsync(InetSocketAddress remotePeer) {
        /***
         * 线程池异步的提交
         */
        threadPoolExecutor.submit(new Runnable() {
            @Override
            public void run() {
                Bootstrap b = new Bootstrap();
                b.group(eventLoopGroup).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new RpcClientInitializer());
                connect(b, remotePeer);
            }
        });
    }

    private void connect(final Bootstrap b, InetSocketAddress remotePeer) {
        //1.真正的建立 连接
        final ChannelFuture channelFuture = b.connect(remotePeer);

        //2.  连接失败 监听  释放资源 进行发起重连操作
        channelFuture.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                //TODO
                log.info("channelFuture.channel close operationComplete remote peer =" + remotePeer);
                channelFuture.channel().eventLoop().schedule(new Runnable() {
                    @Override
                    public void run() {
                        //重新发起连接
                        log.warn(" connect fail to reconnect! ");
                        clearConnected();
                        //递归
                        connect(b, remotePeer);
                    }
                }, 3, TimeUnit.SECONDS);
            }
        });

        //连接成功  回调监听 把我们新连接放入缓存中
        channelFuture.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (future.isSuccess()) {
                    log.info("successfully connect to remote server, remote peer =" + remotePeer);
                    RpcClientHandler handler = future.channel().pipeline().get(RpcClientHandler.class);
                    addHandler(handler);
                }
            }
        });
    }

    /***
     * <h1>添加到RpcClientHandler 到指定的缓存中</h1>
     * concurrentHashMap & connectedHandlerList
     * @param handler
     */
    private void addHandler(RpcClientHandler handler) {
        connectedHandlerList.add(handler);
        InetSocketAddress remotePeer = (InetSocketAddress) handler.getRemotePeer();
        concurrentHashMap.put(remotePeer, handler);
        //signalAvailableHandler  唤醒可用的业务执行器
        signalAvailableHandler();
    }



    //2. 对于连接进来的资源 做个一个缓存(做一个管理) updateConnectedServer

    /***
     * <h1>连接失败  释放资源    清空缓存</h1>
     * 先删除所有的 concurrentHashMap  中的数据
     * 然后在清空 connectedHandlerList 中的数据
     */
    private void clearConnected() {
        for (final RpcClientHandler rpcClientHandler : connectedHandlerList) {
            //通过 RpcClientHandler 找到具体的 remotePeer 从 concurrentHashMap 当中进行移除指定的
            SocketAddress remotePeer = rpcClientHandler.getRemotePeer();
            //缓存的 handler
            RpcClientHandler handler = concurrentHashMap.get(remotePeer);
            if (handler != null) {
                handler.close();
                concurrentHashMap.remove(remotePeer);
            }
        }
        connectedHandlerList.clear();
    }
    /***
     * <h1>唤醒另外一端的线程(阻塞的状态中) 告知有新连接接入</h1>
     */
    private void signalAvailableHandler() {
        reentrantLock.lock();
        try {
            condition.signalAll();
        }finally {
            reentrantLock.unlock();
        }
    }

    /***
     * <h1>等待新连接接入通知方法</h1>
     * @return
     * @throws InterruptedException
     */
    private  boolean waitingForAvailableHandler() throws InterruptedException {
        reentrantLock.lock();
        try {
           return condition.await(this.connectTimeoutMills,TimeUnit.MICROSECONDS);
        }finally {
            reentrantLock.unlock();
        }
    }



    /***
     * <h1>chooseHandler() 选择一个实际的业务处理器</h1>
     * @return RpcClientHandler
     */
    public RpcClientHandler chooseHandler(){

        CopyOnWriteArrayList<RpcClientHandler> handlers = (CopyOnWriteArrayList<RpcClientHandler>)this.connectedHandlerList.clone();
        int size = handlers.size();
        while (isRunning&&size<=0){
            try {
                boolean availableHandler = waitingForAvailableHandler();
                if (availableHandler){
                    //可用
                    handlers=(CopyOnWriteArrayList<RpcClientHandler>)this.connectedHandlerList.clone();
                    size =handlers.size();
                }
            } catch (InterruptedException e) {
                log.error("wating for available node is interrupted!");
                throw new RuntimeException("no connect any servers!",e);
            }
        }
        if(!isRunning){
            return null;
        }

        //最终使用取摸方式取得其中一个业务处理器进行实际的业务处理
        return handlers.get(((handlerIdx.getAndAdd(1) + size) % size));
    }

    /***
     * <h1>停止的方法</h1>
     */
    public void stop(){
        isRunning=false;
        for (int i = 0; i <connectedHandlerList.size() ; i++) {
            RpcClientHandler handler = connectedHandlerList.get(i);
            handler.close();
        }
        //在這裡需要去調用一下喚醒的方法
        signalAvailableHandler();
        //释放资源
        threadPoolExecutor.shutdown();
        eventLoopGroup.shutdownGracefully();
    }

    /***
     * <h1>发起重连的方法 需要把对应的资源进行是释放操作</h1>
     * @param handler
     * @param socketAddress
     */
    public void reconnect(final  RpcClientHandler handler,final SocketAddress socketAddress){
        if (handler!=null){
            handler.close();
            connectedHandlerList.remove(handler);
            concurrentHashMap.remove(socketAddress);
        }
        connectAsync((InetSocketAddress) socketAddress);
    }

}
